package com.ruoyi.system.service.other;
import java.math.BigDecimal;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.ruoyi.common.exception.BusinessException;
import com.ruoyi.common.utils.exception.AjaxException;
import com.ruoyi.system.domain.SysOrder;
import com.ruoyi.system.domain.SysUser;
import com.ruoyi.system.domain.SysUserAccount;
import com.ruoyi.system.domain.SysUserBillcash;
import com.ruoyi.system.mapper.SysUserAccountMapper;
import com.ruoyi.system.mapper.SysUserBillcashMapper;
import com.ruoyi.system.mapper.SysUserMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;


@Component
class StreamMessageListener implements StreamListener<String, MapRecord<String, String, String>>{

    static final Logger LOGGER = LoggerFactory.getLogger(StreamMessageListener.class);

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    private SysUserAccountMapper sysUserAccountMapper;

    @Autowired
    private SysUserBillcashMapper sysUserBillcashMapper;

    @Autowired
    private SysUserMapper sysUserMapper;

    @Autowired
    private SendService sendService;

    @Override
    @Transactional
    public void onMessage(MapRecord<String, String, String> message) {

        // 消息ID
        RecordId messageId = message.getId();

        // 消息的key和value
        Map<String, String> body = message.getValue();

        LOGGER.info("stream message。messageId={}, stream={}, body={}", messageId, message.getStream(), body);

        MessageEntity messageEntity = JSONUtil.toBean(body.get("name"),MessageEntity.class);
        System.out.println("接收到了" + message);
        //支付回调
        if (messageEntity.getKey().equals("pay")){
            SysOrder sysOrder = JSONUtil.toBean(messageEntity.getContent(),SysOrder.class);
            SysUserAccount sysUserAccount = sysUserAccountMapper.selectSysUserAccountById(sysOrder.getUserId());
            if (sysUserAccount==null){
                throw new AjaxException("account invalid");
            }
            BigDecimal oldBalance = sysUserAccount.getBalance();
            BigDecimal newBalance = sysUserAccount.getBalance().add(sysOrder.getOrderLastAmount());
            sysUserAccount.setBalance(newBalance);

            SysUserBillcash sysUserBillcash = new SysUserBillcash();
            sysUserBillcash.setRidCode(sysOrder.getOrderCode());
            sysUserBillcash.setChannelId(sysOrder.getChannelId());
            sysUserBillcash.setChannelName(sysOrder.getChannelName());
            sysUserBillcash.setAmount(oldBalance);
            sysUserBillcash.setChangeAmount(sysOrder.getOrderLastAmount());
            sysUserBillcash.setLastAmount(newBalance);
            SysUser sysUser = sysUserMapper.selectUserById(sysOrder.getUserId());
            if (sysUser==null){
                throw new AjaxException("user error");
            }
            sysUserBillcash.setUserId(sysUser.getUserId());
            sysUserBillcash.setUserCode(sysUser.getUserCode());
            sysUserBillcash.setLoginName(sysUser.getLoginName());
            sysUserBillcash.setBillcashType(0);
            sysUserBillcash.setBillcashRemark("支付完成");
            sysUserBillcash.setCreateTime(new Date());
            sysUserBillcash.setUpdateTime(new Date());
            try {
                sysUserBillcashMapper.insertSysUserBillcash(sysUserBillcash);
            }catch (Exception e){
                throw new BusinessException("system error:billcash");
            }

            //更新账户失败 放回队列
            try {
                sysUserAccountMapper.updateSysUserAccount(sysUserAccount);
            }catch (Exception e){
                throw new BusinessException("system error:account");
            }
        }else {
        }
        // 通过RedisTemplate手动确认消息
        this.stringRedisTemplate.opsForStream().acknowledge("group-1", message);
    }
}


/*    public void run() {
        try {
            ExecutorService executorService = Executors.newCachedThreadPool();
            executorService.execute(() -> {
                MessageEntity message;
                while(true) {
                    message = redisTemplate.opsForList().rightPop("queue", 1000L, TimeUnit.SECONDS);
                    System.out.println("接收到了" + message);
                    //支付回调
                    if (message.getKey().equals(pay)){
                        System.out.println("pay: " + message.getContent());
                        SysOrder sysOrder = JSONUtil.toBean(message.getContent(),SysOrder.class);
                        SysUserAccount sysUserAccount = sysUserAccountMapper.selectSysUserAccountById(sysOrder.getUserId());
                        if (sysUserAccount==null){
                            throw new AjaxException("account invalid");
                        }
                        BigDecimal oldBalance = sysUserAccount.getBalance();
                        BigDecimal newBalance = sysUserAccount.getBalance().add(sysOrder.getOrderLastAmount());
                        sysUserAccount.setBalance(newBalance);

                        SysUserBillcash sysUserBillcash = new SysUserBillcash();
                        sysUserBillcash.setRidCode(sysOrder.getOrderCode());
                        sysUserBillcash.setChannelId(sysOrder.getChannelId());
                        sysUserBillcash.setChannelName(sysOrder.getChannelName());
                        sysUserBillcash.setAmount(oldBalance);
                        sysUserBillcash.setChangeAmount(sysOrder.getOrderLastAmount());
                        sysUserBillcash.setLastAmount(newBalance);
                        SysUser sysUser = sysUserMapper.selectUserById(sysOrder.getUserId());
                        if (sysUser==null){
                            throw new AjaxException("user error");
                        }
                        sysUserBillcash.setUserId(sysUser.getUserId());
                        sysUserBillcash.setUserCode(sysUser.getUserCode());
                        sysUserBillcash.setLoginName(sysUser.getLoginName());
                        sysUserBillcash.setBillcashType(0);
                        sysUserBillcash.setBillcashRemark("支付完成");
                        sysUserBillcash.setCreateTime(new Date());
                        sysUserBillcash.setUpdateTime(new Date());
                        try {
                            sysUserBillcashMapper.insertSysUserBillcash(sysUserBillcash);
                        }catch (Exception e){
                            MessageEntity messageEntity = new MessageEntity();
                            messageEntity.setKey("pay");
                            messageEntity.setContent(JSONUtil.toJsonStr(sysOrder));
                            messageProducerService.sendMeassage(messageEntity);
                        }

                        //更新账户失败 放回队列
                        try {
                            sysUserAccountMapper.updateSysUserAccount(sysUserAccount);
                        }catch (Exception e){
                            MessageEntity messageEntity = new MessageEntity();
                            messageEntity.setKey("pay");
                            messageEntity.setContent(JSONUtil.toJsonStr(sysOrder));
                            messageProducerService.sendMeassage(messageEntity);
                        }
                    }else {
                        System.out.println("paid: " + message.getContent());
                    }
                }
            });
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
    }

    public boolean isFlag() {
        return flag;
    }

    public void setFlag(boolean flag) {
        this.flag = flag;
    }*/

